from django.shortcuts import render
from django.conf import settings
from kafka import KafkaConsumer, TopicPartition
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def consumer_from_kafka(request):
    try:
        logger.info("Creating Kafka consumer...")
        consumer = KafkaConsumer(
            bootstrap_servers=settings.KAFKA_SETTINGS['bootstrap.servers'],
            group_id=settings.KAFKA_SETTINGS['group.id'],
            auto_offset_reset=settings.KAFKA_SETTINGS['auto.offset.reset'],
            request_timeout_ms=settings.KAFKA_SETTINGS['request.timeout.ms'],
            value_deserializer=lambda v: json.loads(v.decode('utf-8'))  # 将消息反序列化为JSON格式
        )
        logger.info(f"Kafka consumer created with group_id: {settings.KAFKA_SETTINGS['group.id']}")

        # 获取主题的分区信息
        topic = 'SSJOM_PROJ-LOGS_COLLECTION'
        logger.info(f"Fetching partitions for topic: {topic}")
        # partitions_for_topic 方法获取该主题的所有分区的编号列表。
        partitions = consumer.partitions_for_topic(topic)
        logger.info(f"Topic {topic} has {len(partitions)} partitions.")

        # 获取最新的偏移量
        """
            TopicPartition 创建一个包含主题和分区号的元组。
            consumer.assign([tp]) 将消费者分配到这个特定的分区。
            consumer.seek_to_end(tp) 将消费者在指定分区的读取位置移动到末尾。
            consumer.position(tp) 获取当前消费者在指定分区的位置（偏移量），并将其存储在 latest_offsets 字典中。
        """
        latest_offsets = {}
        for partition in partitions:
            tp = TopicPartition(topic, partition)
            logger.info(f"Assigning consumer to partition: {tp}")
            consumer.assign([tp])
            consumer.seek_to_end(tp)
            latest_offsets[tp] = consumer.position(tp)
            logger.info(f"Latest offset for partition {tp} is {latest_offsets[tp]}")

        # 消费最近的消息
        """
            consumer.seek(tp, max(0, offset - 100)) 将消费者在指定分区的位置移动到距离末尾 100 条消息之前的位置。如果偏移量小于 100，
            则移动到最开始的位置（offset - 100 可能会是负数，所以使用 max 函数确保不会小于 0）。
            for message in consumer: 开始从当前位置消费消息。
            recent_messages.append(message.value) 将消息的值（不包括元数据）添加到 recent_messages 列表中。
            如果 recent_messages 列表中的消息数量达到 100 条，就停止从当前分区读取更多消息。
            如果已经收集了 100 条消息，就停止遍历所有分区。
        """
        recent_messages = []
        for tp, offset in latest_offsets.items():
            logger.info(f"Seeking to offset {max(0, offset - 10)} for partition {tp}")
            consumer.seek(tp, max(0, offset - 10))
            for message in consumer:
                recent_messages.append(message.value)
                logger.info(f"Consumed message from partition {tp}: {message.value}")
                if len(recent_messages) >= 10:
                    break
            if len(recent_messages) >= 10:
                break

        logger.info(f"Total messages consumed: {len(recent_messages)}")
        return render(request, 'extensions/consumerList.html', {'consumerList': recent_messages})

    except Exception as e:
        logger.error(f"Error consuming messages from Kafka: {e}")
        return render(request, 'extensions/consumerList.html', {'consumerList': []})